mod data_table;
mod inrc_join_table;
mod strconv;
mod tmp_table;
mod column_datas;
use std::{
    fs::File, io::{BufRead, BufReader}, mem::swap, path::PathBuf
};
use crate::{
    assess_proc,
    base::{
        get_imp_config, get_sys_date, import_begin_log, import_finish_log, import_is_finish,
        load_feqp, source_begin_import, source_finish_import, table_has_data_with_date,
        CodedFormat, Dependent, Error, Frequency, ImportType, Result, SourceType, Table,
    },
    ftp::FtpDownload,
};
use chrono::{Datelike, Days, NaiveDate};
pub use data_table::*;
use encoding_rs::GBK;
use encoding_rs_io::DecodeReaderBytesBuilder;
use memmap2::Mmap;

pub(crate) async fn import_one_source(
    date: NaiveDate,
    source_type: SourceType,
    is_again: bool,
) -> Result<()> {
    let ic = get_imp_config(source_type);
    if ic.frequency == Frequency::Monthly {
        let nd = date.checked_add_days(Days::new(1)).unwrap();
        if nd.day() != 1 {
            return Ok(());
        }
    }
    let file_name_prefix = ic.file_name_prefix.as_str();
    let file_ext_name = ic.file_ext_name.as_str();
    let confirm_file_ext_name = ic.confirm_file_ext_name.as_str();
    let coded_format = &ic.coded_format;
    let mut ftp = FtpDownload::from_conf(&ic.ftp)?;
    let count = load_feqp(&date).await?;
    if source_type != SourceType::ODS && count == 0 {
        let date_str = date.format("%Y%m%d").to_string();
        return Err(anyhow::anyhow!("{date_str}日的汇率表为空，请联系管理员").into());
    }
    source_begin_import(&date, &source_type).await?;
    for tconf in &ic.tables {
        import_one_file(
            date,
            source_type,
            file_name_prefix,
            file_ext_name,
            confirm_file_ext_name,
            ic.column_separator,
            coded_format,
            tconf,
            &mut ftp,
            is_again,
        )
        .await?;
    }
    source_finish_import(&date, &source_type, false).await?;
    if source_type == SourceType::EDW {
        assess_proc(None).await?;
    }
    Ok(())
}

pub(crate) async fn import_one_file(
    date: NaiveDate,
    source_type: SourceType,
    file_name_prefix: &str,
    file_ext_name: &str,
    confirm_file_ext_name: &str,
    column_separator: u8,
    coded_format: &CodedFormat,
    tconf: &Table,
    ftp: &mut FtpDownload,
    is_again: bool,
) -> Result<()> {
    if !is_again {
        let is_finish = import_is_finish(&date, &tconf.file_name, &source_type).await?;
        if is_finish {
            log::warn!("文件{}已导入,已跳过", tconf.file_name);
            return Ok(());
        }
    } else {
        for dep in tconf.dependents.iter() {
            if dep.is_along {
                let ic = get_imp_config(dep.source);
                let file_name_prefix = ic.file_name_prefix.as_str();
                let file_ext_name = ic.file_ext_name.as_str();
                let confirm_file_ext_name = ic.confirm_file_ext_name.as_str();
                let coded_format = &ic.coded_format;
                let tconf = ic
                    .tables
                    .iter()
                    .find(|t| t.file_name == dep.file_name)
                    .ok_or(anyhow::anyhow!(
                        "在数据源配置{}中未找到文件{}的配置",
                        source_type,
                        dep.file_name
                    ))?;
                let future = Box::pin(import_one_file(
                    date.clone(),
                    dep.source,
                    file_name_prefix,
                    file_ext_name,
                    confirm_file_ext_name,
                    ic.column_separator,
                    coded_format,
                    tconf,
                    ftp,
                    is_again,
                ));
                future.await?;
            }
        }
    }
    let date_str = date.format("%Y%m%d").to_string();
    log::info!("开始导入{source_type}{date_str}文件:{}",&tconf.file_name);
    let is_deps_ok = is_dependents_ok(&date, &tconf.dependents).await;
    if let Err(err) = is_deps_ok {
        if tconf.is_must {
            return Err(Error::from(anyhow::anyhow!("依赖文件:{}", err.to_string())));
        }
        log::warn!("依赖文件:{},已跳过", err.to_string());
    }
    if tconf.import_type == ImportType::DayTrls || tconf.import_type == ImportType::MonthTrls {
        let has_data = table_has_data_with_date(&tconf.table_name, &date).await?;
        if has_data {
            log::warn!("表{}已经存在{date_str}的数据,已跳过", &tconf.table_name);
            return Ok(());
        }
    }
    import_begin_log(&date, &tconf.file_name, &tconf.table_name, &source_type).await?;
    let file_name = format!(
        "{}{}{}_{}.{}",
        file_name_prefix, tconf.file_name, tconf.file_name_suffix, date_str, confirm_file_ext_name
    );
    let file_path = ftp.download(&file_name);
    if let Err(err) = file_path {
        if tconf.is_must {
            return Err(
                anyhow::anyhow!("文件{}下载失败: {}", tconf.file_name, err.to_string()).into(),
            );
        } else {
            log::warn!("文件{}下载失败,已跳过", tconf.file_name.as_str());
            return Ok(());
        }
    }
    let file_sn=if tconf.file_sn.is_empty(){
        "".to_owned()
    }else{
        format!("_{}",tconf.file_sn)
    };
    let file_name = format!(
        "{}{}{}_{date_str}{file_sn}.{file_ext_name}",
        file_name_prefix,
        tconf.file_name,
        tconf.file_name_suffix,
    );
    let file_path = ftp.download(&file_name);
    if let Err(err) = file_path {
        if tconf.is_must {
            return Err(
                anyhow::anyhow!("文件{}下载失败: {}", tconf.file_name, err.to_string()).into(),
            );
        } else {
            log::warn!("文件{}下载失败,已跳过", tconf.file_name.as_str());
            return Ok(());
        }
    }
    let file_path = file_path?;
    let sign=column_separator as char;
    let mut dt = DataTable::from(source_type,tconf,sign).await?;
    //导入前执行
    dt.before_import().await?;
    let count = match coded_format {
        CodedFormat::UTF8 => {
            utf8_read(file_path, &mut dt, tconf).await?
        }
        CodedFormat::GBK => {
            gbk_read(file_path, &mut dt, tconf).await?
        }
    };
    if count == 0 {
        if !tconf.can_blank&&tconf.is_must {
            return Err(anyhow::anyhow!("文件{}无数据", tconf.file_name.as_str()).into());
        } else {
            log::warn!("文件{}无数据,已跳过", tconf.file_name.as_str());
            return Ok(());
        }
    }
    //保存缓冲区数据
    dt.flush().await?;
    log::info!("导入{source_type}{date_str}文件:{}完成，共导入{count}行数据",&tconf.file_name);
    //导入后执行
    dt.after_import(&date).await?;
    import_finish_log(&date, &tconf.file_name, count, &source_type).await?;
    log::info!("{source_type}{date_str}文件{}数据处理完成",&tconf.file_name);
    Ok(())
}

async fn utf8_read(
    file_path: PathBuf,
    dt: &mut DataTable<'_>,
    tconf: &Table,
) -> Result<u64> {
    let file = File::open(file_path)?;
    let mmap = unsafe { Mmap::map(&file)? };
    let buff = BufReader::new(&mmap[..]);
    let mut count = 0u64;
    let mut batch = Vec::new();
    for (i, res) in buff.lines().enumerate() {
        if i < tconf.start_row_no as usize {
            continue;
        }
        let line_str = res.or_else(|e| {
            Err(Error::from_txt(
                tconf.file_name.as_str(),
                i + 1,
                0,
                format!("读取文件行失败: {}", e.to_string()),
            ))
        })?;
        batch.push(line_str);
        if batch.len() >= 1000 {
            count += batch.len() as u64;
            let mut batch2 = Vec::new();
            swap(&mut batch, &mut batch2);
            dt.insert_rows(batch2).await?; // 批量插入
            batch.clear();
        }
    }
    if !batch.is_empty() {
        count += batch.len() as u64;
        dt.insert_rows(batch).await?;
    }

    Ok(count)
}

async fn gbk_read(
    file_path: PathBuf,
    dt: &mut DataTable<'_>,
    tconf: &Table,
) -> Result<u64> {
    let file = File::open(file_path)?;
    let mmap = unsafe { Mmap::map(&file)? };
    let decoder = DecodeReaderBytesBuilder::new()
        .encoding(Some(GBK))
        .build(&mmap[..]);
    let reader = BufReader::new(decoder);
    let mut count = 0u64;
    let mut batch = Vec::new();

    for (i, res) in reader.lines().enumerate() {
        if i < tconf.start_row_no as usize {
            continue;
        }
        let line_str = res.or_else(|e| {
            Err(Error::from_txt(
                tconf.file_name.as_str(),
                i + 1,
                0,
                format!("读取文件行失败: {}", e.to_string()),
            ))
        })?;
        batch.push(line_str);

        if batch.len() >= 1000 {
            count += batch.len() as u64;
            let mut batch2 = Vec::new();
            swap(&mut batch, &mut batch2);
            dt.insert_rows(batch2).await?; // 批量插入
        }
    }

    if !batch.is_empty() {
        count += batch.len() as u64;
        dt.insert_rows(batch).await?;
    }

    Ok(count)
}

async fn is_dependents_ok(date: &NaiveDate, dependents: &Vec<Dependent>) -> anyhow::Result<()> {
    for dep in dependents.iter() {
        let date2 = get_sys_date(dep.source).await?;
        if date2 >= *date {
            return Ok(());
        }
        if !import_is_finish(date, &dep.file_name, &dep.source).await? {
            return Err(anyhow::anyhow!(
                "数据源:{}文件:{}日期:{}没有导入",
                dep.source,
                dep.file_name,
                date
            ));
        }
    }
    Ok(())
}
